//queue.c

#include <errno.h>
#include <pthread.h>
#include "queue.h"
#include "global.h"
#include "logger.h"
#include "shared_func.h"

static struct fnio_queue_info g_free_queue;

static struct fnio_buff_node *g_mpool = NULL;

static int _queue_push_task(struct fnio_queue_info *pQueue, \
		struct fnio_buff_node *pNode);
static struct fnio_buff_node *_queue_pop_task(struct fnio_queue_info *pQueue);
static int _fnio_queue_count(struct fnio_queue_info *pQueue);

int fnio_queue_init()
{
	struct fnio_buff_node *pNode;
	struct fnio_buff_node *pEnd;
	int alloc_size;
	int result;

	if ((result=init_pthread_lock(&(g_free_queue.lock))) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"init_pthread_lock fail, program exit!", __LINE__);
		return result;
	}

	alloc_size = sizeof(struct fnio_buff_node) * g_max_connections;
	g_mpool = (struct fnio_buff_node *)malloc(alloc_size);
	if (g_mpool == NULL)
	{
		logError("file: "__FILE__", line: %d, " \
			"malloc %d bytes fail", \
			__LINE__, alloc_size);
		return errno != 0 ? errno : ENOMEM;
	}

	memset(g_mpool, 0, alloc_size);
	pEnd = g_mpool + g_max_connections;
	for (pNode=g_mpool; pNode<pEnd; pNode++)
	{
		pNode->size = g_min_buff_size;
		pNode->data = malloc(pNode->size);
		if (pNode->data == NULL)
		{
			fnio_queue_destroy();

			logError("file: "__FILE__", line: %d, " \
				"malloc %d bytes fail", \
				__LINE__, pNode->size);
			return errno != 0 ? errno : ENOMEM;
		}
	}

	g_free_queue.head = g_mpool;
	g_free_queue.tail = pEnd - 1;
	for (pNode=g_mpool; pNode<g_free_queue.tail; pNode++)
	{
		pNode->next = pNode + 1;
	}
	g_free_queue.tail->next = NULL;

	return 0;
}

void fnio_queue_destroy()
{
	struct fnio_buff_node *pNode;
	struct fnio_buff_node *pEnd;

	if (g_mpool == NULL)
	{
		return;
	}

	pEnd = g_mpool + g_max_connections;
	for (pNode=g_mpool; pNode<pEnd; pNode++)
	{
		if (pNode->data != NULL)
		{
			free(pNode->data);
			pNode->data = NULL;
		}
	}

	free(g_mpool);
	g_mpool = NULL;

	pthread_mutex_destroy(&(g_free_queue.lock));
}

struct fnio_buff_node *free_queue_pop()
{
	return _queue_pop_task(&g_free_queue);;
}

int free_queue_push(struct fnio_buff_node *pNode)
{
	pNode->length = 0;
	pNode->offset = 0;

	memset(pNode->data, 0, pNode->size);
	return _queue_push_task(&g_free_queue, pNode);
}

int free_queue_count()
{
	return _fnio_queue_count(&g_free_queue);
}

static int _queue_push_task(struct fnio_queue_info *pQueue, \
		struct fnio_buff_node *pNode)
{
	int result;

	if ((result=pthread_mutex_lock(&(pQueue->lock))) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"call pthread_mutex_lock fail, " \
			"errno: %d, error info: %s", \
			__LINE__, result, strerror(result));
		return result;
	}

	pNode->next = NULL;
	if (pQueue->tail == NULL)
	{
		pQueue->head = pNode;
	}
	else
	{
		pQueue->tail->next = pNode;
	}
	pQueue->tail = pNode;

	if ((result=pthread_mutex_unlock(&(pQueue->lock))) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"call pthread_mutex_unlock fail, " \
			"errno: %d, error info: %s", \
			__LINE__, result, strerror(result));
	}

	return 0;
}

static struct fnio_buff_node *_queue_pop_task(struct fnio_queue_info *pQueue)
{
	struct fnio_buff_node *pNode;
	int result;

	if ((result=pthread_mutex_lock(&(pQueue->lock))) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"call pthread_mutex_lock fail, " \
			"errno: %d, error info: %s", \
			__LINE__, result, strerror(result));
		return NULL;
	}

	pNode = pQueue->head;
	if (pNode != NULL)
	{
		pQueue->head = pNode->next;
		if (pQueue->head == NULL)
		{
			pQueue->tail = NULL;
		}
	}

	if ((result=pthread_mutex_unlock(&(pQueue->lock))) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"call pthread_mutex_unlock fail, " \
			"errno: %d, error info: %s", \
			__LINE__, result, strerror(result));
	}

	return pNode;
}

static int _fnio_queue_count(struct fnio_queue_info *pQueue)
{
	struct fnio_buff_node *pNode;
	int count;
	int result;

	if ((result=pthread_mutex_lock(&(pQueue->lock))) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"call pthread_mutex_lock fail, " \
			"errno: %d, error info: %s", \
			__LINE__, result, strerror(result));
		return 0;
	}

	count = 0;
	pNode = pQueue->head;
	while (pNode != NULL)
	{
		pNode = pNode->next;
		count++;
	}

	if ((result=pthread_mutex_unlock(&(pQueue->lock))) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"call pthread_mutex_unlock fail, " \
			"errno: %d, error info: %s", \
			__LINE__, result, strerror(result));
	}

	return count;
}


int fnio_malloc_send_buff(struct fnio_send_buff *pSendBuff)
{
	pSendBuff->malloc_size = g_min_buff_size;
	pSendBuff->send_buff = (char *)malloc(pSendBuff->malloc_size);
	if (pSendBuff->send_buff == NULL)
	{
		logError("file: "__FILE__", line: %d, " \
			"malloc %d bytes fail, " \
			"errno: %d, error info: %s", \
			__LINE__, pSendBuff->malloc_size, \
			errno, strerror(errno));
		return errno != 0 ? errno : ENOMEM;
	}

	pSendBuff->max_used_bytes = pSendBuff->malloc_size;
	pSendBuff->buff_end = pSendBuff->send_buff + pSendBuff->malloc_size;
	pSendBuff->send_start = pSendBuff->send_buff;
	pSendBuff->free_start = pSendBuff->send_buff;

	return 0;
}

static int fnio_remalloc_send_buff(struct fnio_send_buff *pSendBuff, \
		const int new_size)
{
	char *p;
	int bytes1;
	int bytes2;

	p = (char *)malloc(new_size);
	if (p == NULL)
	{
		logError("file: "__FILE__", line: %d, " \
			"malloc %d bytes fail, " \
			"errno: %d, error info: %s", \
			__LINE__, new_size, errno, strerror(errno));
		return errno != 0 ? errno : ENOMEM;
	}

	pSendBuff->malloc_size = new_size;
	pSendBuff->max_used_bytes = pSendBuff->malloc_size;
	pSendBuff->keep_count = 0;
	if (pSendBuff->free_start > pSendBuff->send_start)
	{
		memcpy(p, pSendBuff->send_start, pSendBuff->send_bytes);
	}
	else if (pSendBuff->send_bytes > 0)
	{
		bytes1 = pSendBuff->buff_end - pSendBuff->send_start;
		bytes2 = pSendBuff->send_bytes - bytes1;
		memcpy(p, pSendBuff->send_start, bytes1);
		memcpy(p + bytes1, pSendBuff->send_buff, bytes2);
	}

	free(pSendBuff->send_buff);
	pSendBuff->send_buff = p;
	pSendBuff->buff_end = pSendBuff->send_buff + \
			      pSendBuff->malloc_size;
	pSendBuff->send_start = pSendBuff->send_buff;
	pSendBuff->free_start = pSendBuff->send_buff + \
				pSendBuff->send_bytes;

	return 0;
}

int fnio_push_to_send_buff(struct fnio_send_buff *pSendBuff, \
		const char *data1, const int data_len1, \
		const char *data2, const int data_len2)
{
	int free_bytes;
	int current_bytes;
	int new_size;
	int total_bytes;
	int bytes1;
	int bytes2;
	int result;

	current_bytes = data_len1 + data_len2;
	total_bytes = pSendBuff->send_bytes + current_bytes;
	if (total_bytes > g_max_buff_size)
	{
		return ENOSPC;
	}

	if (total_bytes > pSendBuff->malloc_size)
	{
		new_size = pSendBuff->malloc_size * 2;
		while (total_bytes > new_size)
		{
			new_size *= 2;
		}

		if (new_size > g_max_buff_size)
		{
			new_size = g_max_buff_size;
		}

		if ((result=fnio_remalloc_send_buff(pSendBuff, new_size)) != 0)
		{
			return result;
		}
	}

	if (pSendBuff->free_start > pSendBuff->send_start)
	{
		free_bytes = (pSendBuff->send_start - \
				pSendBuff->send_buff) + \
				(pSendBuff->buff_end - \
				pSendBuff->free_start);
	}
	else
	{
		if (pSendBuff->send_bytes == 0)
		{
			free_bytes = pSendBuff->malloc_size;
		}
		else
		{
			free_bytes = pSendBuff->send_start - \
				     pSendBuff->free_start;
		}
	}

	if (current_bytes > free_bytes)
	{
		return ENOSPC;
	}

	if ((pSendBuff->buff_end - pSendBuff->free_start) <= current_bytes)
	{
		bytes1 = pSendBuff->buff_end - \
                	pSendBuff->free_start;
		if (bytes1 < data_len1)
		{
			bytes2 = data_len1 - bytes1;
			memcpy(pSendBuff->free_start, data1, bytes1);
			pSendBuff->free_start = pSendBuff->send_buff;
			memcpy(pSendBuff->free_start, data1+bytes1, bytes2);
			pSendBuff->free_start += bytes2;
			memcpy(pSendBuff->free_start, data2, data_len2);
			pSendBuff->free_start += data_len2;
		}
		else
		{
			bytes1 -= data_len1;
			bytes2 = data_len2 - bytes1;

			if (data_len1 > 0)
			{
				memcpy(pSendBuff->free_start, data1, data_len1);
				pSendBuff->free_start += data_len1;
			}
			memcpy(pSendBuff->free_start, data2, bytes1);
			pSendBuff->free_start = pSendBuff->send_buff;
			memcpy(pSendBuff->free_start, data2+bytes1, bytes2);
			pSendBuff->free_start += bytes2;
		}
	}
	else
	{
		if (data_len1 > 0)
		{
			memcpy(pSendBuff->free_start, data1, data_len1);
			pSendBuff->free_start += data_len1;
		}
		memcpy(pSendBuff->free_start, data2, data_len2);
		pSendBuff->free_start += data_len2;
	}

	pSendBuff->send_bytes += current_bytes;

	if (pSendBuff->send_bytes > pSendBuff->max_used_bytes)
	{
		do
		{
			pSendBuff->max_used_bytes *= 2;
		}
		while (pSendBuff->send_bytes > pSendBuff->max_used_bytes);

		if (pSendBuff->max_used_bytes > g_max_buff_size)
		{
			pSendBuff->max_used_bytes = g_max_buff_size;
		}

		pSendBuff->keep_count = 0;  //re-count
	}

	pSendBuff->keep_count++;
	if (pSendBuff->keep_count == 100)
	{
		if (pSendBuff->malloc_size > pSendBuff->max_used_bytes)
		{
			if ((result=fnio_remalloc_send_buff(pSendBuff,
					pSendBuff->max_used_bytes)) != 0)
			{
				return result;
			}
		}
		else
		{
			pSendBuff->max_used_bytes = g_min_buff_size;
			pSendBuff->keep_count = 0;  //re-count
		}
	}

	return 0;
}

